package cn.doitedu.benchmark;

import org.apache.commons.lang3.RandomUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class TrafficOverviewTest {

    public static void main(String[] args) throws InterruptedException {

        for(int i=0;i<40;i++){
            new Thread(new Task()).start();
            Thread.sleep(2000);
        }

    }


    public static class Task implements Runnable{

        @Override
        public void run() {
            Properties properties = new Properties();
            properties.setProperty("bootstrap.servers","doitedu:9092");
            properties.setProperty("key.serializer", StringSerializer.class.getName());
            properties.setProperty("value.serializer", StringSerializer.class.getName());


            KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

            String json = "{\"user_id\":$uid,\"username\":\"windy\",\"session_id\":\"$sid\",\"event_id\":\"$eid\",\"action_time\":$ts,\"lat\":38.089969323508726,\"lng\":114.35731900345093,\"release_channel\":\"华为应用市场\",\"device_type\":\"mi8\",\"properties\":{\"url\":\"$pid\",\"item_id\":1,\"keyword\":\"usb 移动固态\",\"favorite_id\":\"sc01\"},\"register_phone\":\"18061581848\",\"user_status\":1,\"register_time\":\"2018-08-03 16:46:38\",\"register_gender\":1,\"register_birthday\":\"2002-03-06\",\"register_province\":\"上海\",\"register_city\":\"上海\",\"register_job\":\"程序员\",\"register_source_type\":1,\"province\":\"河北省\",\"city\":\"石家庄市\",\"region\":\"鹿泉区\",\"page_type\":\"文章页\",\"page_service\":\"内容服务\"}\n";

            for(int i = 0; i< 10000000; i++){

                String uid = RandomUtils.nextInt(1,10000000)+"";
                String sid = RandomUtils.nextInt(1000000,10000000)+"";
                String eid = "page_load";
                if(RandomUtils.nextInt(1,10)>8){
                    eid = "event-"+RandomUtils.nextInt(1,100);
                }
                String ts = System.currentTimeMillis()+"";


                String tmp1 = json.replace("$uid", uid);
                String tmp2 = tmp1.replace("$sid", sid);
                String tmp3 = tmp2.replace("$eid", eid);
                String tmp4 = tmp3.replace("$ts", ts);
                String tmp5 = tmp4.replace("$pid", "/page/" + RandomUtils.nextInt(1, 10000));


                ProducerRecord<String, String> record = new ProducerRecord<>("dwd-events", tmp5);
                producer.send(record);
                producer.flush();
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }


}
